package cn.piesat.scanning.business.common.algorithm.processAlgorithm;

import cn.piesat.scanning.dto.HtScanningMetadataDTO;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import cn.piesat.scanning.business.dq1043.flow.FlowTask;
import cn.piesat.scanning.dto.DbDmsSchedulerTaskFileRecordDTO;
import cn.piesat.scanning.dto.DbDmsSchedulerTaskFileRecordFlowDTO;
import cn.piesat.scanning.dto.HtScheduleJobDTO;
import cn.piesat.scanning.service.*;
import cn.piesat.scanning.utils.RedisUtil;
import cn.piesat.scanning.vo.flow.FlowVO;
import cn.piesat.scanning.vo.flow.LineVO;
import cn.piesat.scanning.vo.flow.NodeVO;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;

import java.util.*;
import java.util.stream.Collectors;

/**
 * 回调算法监听
 * @author xhl
 */
@Component
public class ReSendMessageListener {

    @Autowired
    private FlowAlgorithmParamsService flowAlgorithmParamsService;
    @Autowired
    private HtScheduleJobService htScheduleJobService;
    @Autowired
    private DbDmsSchedulerTaskFileRecordService dbDmsSchedulerTaskFileRecordService;
    @Autowired
    private DbDmsSchedulerTaskFileRecordFlowService dbDmsSchedulerTaskFileRecordFlowService;
    @Autowired
    private HtScanningMetadataService htScanningMetadataService;
    @Autowired
    private FlowTask flowTask;

//    @KafkaListener(id = "DMS_RE_SEND_CONSUMER",topics = "DMS_RE_SEND_EXEC")
//    public void listener(ConsumerRecord<String, String> record, Consumer<String, String> consumer) {
//        Jedis jedis = RedisUtil.getJedis();
//        try {
//            if (jedis != null) {
//                String value = record.value();
//                JSONObject jsonObject = JSON.parseObject(value);
//                String viewNumber = jsonObject.getString("viewNumber");
//                String algorithmId = jsonObject.getString("algorithmId");
//                Integer status = jsonObject.getInteger("status");
//                String taskId = jsonObject.getString("taskId");
//                String endTime = jsonObject.getString("endTime");
//                String fileName = jsonObject.getString("mainFileName");
//                //判断当前节点是否执行完毕
//                if (StringUtils.isBlank(fileName)) {
//                    fileName = "noMainFile";
//                }
//                HtScheduleJobDTO taskDTO = htScheduleJobService.findById(taskId);
//                String recordCountKey = String.format("recordCount:%s:%s:%s:%s",taskId,viewNumber,algorithmId,fileName);
//                jedis.hset(recordCountKey,"state",status + "");
//                if (status == 7) {
//                    //TODO 重做逻辑
//                    HtScanningMetadataDTO metadata = htScanningMetadataService.findByFileName(fileName);
//                    flowTask.singleFileRedo(algorithmId,taskDTO,metadata);
//                } else {
//                    boolean flag = true;
//                    recordCountKey = String.format("recordCount:%s:%s:%s:*",taskId,viewNumber,algorithmId);
//                    Set<String> keys = jedis.keys(recordCountKey);
//                    for (String key : keys) {
//                        int state = Integer.parseInt(jedis.hget(key, "state"));
//                        if (state == 0 || state == 1 || state == 7) {
//                            flag = false;
//                            break;
//                        }
//                    }
//                    if (flag) {
//                        //查找下级算法参数信息
//                        String flowJson = jedis.get(String.format("flow:%s:%s",taskId,viewNumber));
//                        List<FlowAlgorithmParam> nextParams = findNextParams(flowJson,algorithmId,taskDTO.getFlowId());
//                        if (nextParams == null) {
//                            //说明该节点的下个节点为结束节点
//                            String nodeCountKey = String.format("node:%s:%s",taskId,viewNumber);
//                            String nodeCount = jedis.get(nodeCountKey);
//                            long nodeTtl = jedis.ttl(nodeCountKey);
//                            Long count = 1L;
//                            if (StringUtils.isBlank(nodeCount)) {
//                                jedis.set(nodeCountKey,"1");
//                                nodeTtl = 3600 * 48;
//                            } else {
//                                count = jedis.incr(nodeCountKey);
//                            }
//                            jedis.expire(nodeCountKey,(int)nodeTtl);
//                            FlowVO flowVO = JSON.parseObject(flowJson, FlowVO.class);
//                            List<LineVO> lineList = flowVO.getLineList();
//                            long endCount = lineList.stream()
//                                    .filter(line -> "nodeB".equals(line.getTo()))
//                                    .count();
//                            //如果结束节点的所有上级节点都执行完毕，则修改记录状态，并存储流程信息
//                            if (count >= endCount) {
//                                DbDmsSchedulerTaskFileRecordDTO taskRecord = dbDmsSchedulerTaskFileRecordService.findByTaskIdAndFileName(taskId, viewNumber);
//                                taskRecord.setState(2);
//                                Date endDate = new Date();
//                                if (StringUtils.isNotBlank(endTime)) {
//                                    endDate = new Date(Long.parseLong(endTime));
//                                }
//                                taskRecord.setEndTime(endDate);
//                                String recordKey = String.format("record:%s:%s",taskId,viewNumber);
//                                long recordTtl = jedis.ttl(recordKey);
//                                jedis.set(recordKey,JSON.toJSONString(taskRecord));
//                                jedis.expire(recordKey,(int)recordTtl);
//                                dbDmsSchedulerTaskFileRecordService.save(taskRecord);
//                                DbDmsSchedulerTaskFileRecordFlowDTO flowDTO = new DbDmsSchedulerTaskFileRecordFlowDTO();
//                                flowDTO.setFlowInfo(flowJson);
//                                flowDTO.setRid(taskRecord.getId());
//                                dbDmsSchedulerTaskFileRecordFlowService.save(flowDTO);
//                            }
//                        } else {
//                            //TODO 调用实现的任务类中的扫描文件方法
//                            List<String> viewNumberList = new ArrayList<>();
//                            viewNumberList.add(viewNumber);
//                            flowTask.parseRuleAndFindData(nextParams,taskDTO,viewNumberList);
//                        }
//                    }
//                }
//            }
//        } finally {
//            RedisUtil.close(jedis);
//            consumer.commitAsync();
//        }
//    }

    /**
     * 查找下级算法参数信息
     * @param flowJson 当前流程
     * @param algorithmId 当前算法ID
     * @return
     */
    private List<FlowAlgorithmParam> findNextParams(String flowJson, String algorithmId, String flowId) {
        FlowVO flowVO = JSON.parseObject(flowJson, FlowVO.class);
        List<LineVO> lineList = flowVO.getLineList();
        List<NodeVO> nodeList = flowVO.getNodeList();
        Map<String, String> nodeHash = nodeList.stream()
                .filter(nodeVO -> StringUtils.isNotBlank(nodeVO.getNodeId()))
                .collect(Collectors.toMap(NodeVO::getId, NodeVO::getNodeId, (k1, k2) -> k2));
        List<String> nextAlgIdList = new ArrayList<>();
        for (NodeVO nodeVO : nodeList) {
            if (StringUtils.isNotBlank(nodeVO.getNodeId()) && nodeVO.getNodeId().equals(algorithmId)) {
                for (LineVO lineVO : lineList) {
                    if (lineVO.getFrom().equals(nodeVO.getId())) {
                        if (lineVO.getTo().equals("nodeB")){
                            return null;
                        }
                        nextAlgIdList.add(nodeHash.get(lineVO.getTo()));
                    }
                }
                break;
            }
        }
        return nextAlgIdList.stream()
                .map(algId -> {
                    FlowAlgorithmParam param = new FlowAlgorithmParam();
                    param.setAlgorithmId(algId);
                    if (StringUtils.isNotBlank(algId)) {
                        Map<String, Object> inputParam = flowAlgorithmParamsService.findDynamicParams(flowId, algId);
                        param.setFlowInputParam(inputParam);
                        Map<String, Object> ruleParam = flowAlgorithmParamsService.findFixedParams(flowId, algId);
                        param.setFlowRuleParam(ruleParam);
                    }
                    return param;
                }).collect(Collectors.toList());
    }
}
